package com.dagger.threads;

import java.util.Date;
import java.util.Iterator;
import java.util.Set;

import org.springframework.data.redis.core.RedisTemplate;

import com.dagger.config.MsgConfig;
import com.dagger.constant.MessageConstants;
import com.dagger.constants.LocalErrorCode;
import com.dagger.utils.AbstractMsgUtil;
import com.dagger.utils.SpringUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * 延时消息消费者
 * @author dagger
 *
 */
@Slf4j
public class DelayConsumerThread extends Thread {
	private String bussinessType;
	private String bussinessId;
	private String queueKey;
	private RedisTemplate redisTemplate;
	// 消息过期时间，过期后删除
	private long msgExpireMillisecond;
	// 消息延迟时间
	private long msgDelayMillisecond;

	public DelayConsumerThread(String bussinessType, String bussinessId) {
		this.bussinessType = bussinessType;
		this.bussinessId = bussinessId;
		String msgKey = MessageConstants.getMsgKey(bussinessType, bussinessId);
//		this.delayMsgKey=PushApiConstants.getDelayMsgKey(msgKey);

//		this.delayKey = PushApiConstants.getDelayKey(msgKey);

		this.redisTemplate = SpringUtils.getBean(RedisTemplate.class);

		msgDelayMillisecond = MsgConfig.getInstance().getDelayMillisecond(msgKey);

		if (msgDelayMillisecond <= 0) {
			throw new RuntimeException(LocalErrorCode.MSG_NOT_DELAY);
		}

		this.queueKey = MessageConstants.getDelayQueueKey(msgKey);
		msgExpireMillisecond = msgDelayMillisecond * 3;
	}

	@Override
	public void run() {
		Double queueMaxScore = 0d;// 队列最大分数（最近的时间）
		Double rangeMinScore = 0d;// 取值范围最小的分数
		Double rangeMaxScore = 0d;// 取值范围最大的分数
		Long lastTime = 0l;// 上一次时间

		while (true) {
			try {
				if (!AbstractMsgUtil.needMsg(bussinessType, bussinessId)) {
					return;
				}

				// 获取最大的分数（最近的一条消息时间）
				Set<String> maxSet = redisTemplate.opsForZSet().reverseRange(queueKey, 0, 0);
				if (maxSet.isEmpty()) {// 队列空，暂停一会儿后继续
					sleep();
					continue;
				}

				queueMaxScore = redisTemplate.opsForZSet().score(queueKey, maxSet.iterator().next());

				rangeMaxScore = queueMaxScore - msgDelayMillisecond;// 队列中可取的数据最大分数，如果没有新数据，此值不会再变
				if (rangeMinScore == 0) {
					// 第一次取数据时，取可用的数据之前的0.5秒缓冲数据
					rangeMinScore = rangeMaxScore - 500;
					lastTime = new Date().getTime();// 当前时间
					log.info("msg range:{}~{}-{}={}", rangeMinScore, queueMaxScore, msgDelayMillisecond, rangeMaxScore);
				} else {
					long time = new Date().getTime() - lastTime;// 两次时间差
					double tmpTime = rangeMinScore + time;// 理论上可以查询到的数据范围
					if (rangeMaxScore < tmpTime) {
						rangeMaxScore = tmpTime;// 解决没有新数据时无法获取最后几条数据的问题
						log.info("msg range:{}~{}+{}={}", rangeMinScore, rangeMinScore, time, rangeMaxScore);
					} else {
						log.info("msg range:{}~{}-{}={}", rangeMinScore, queueMaxScore, msgDelayMillisecond,
								rangeMaxScore);
					}
				}

				// 删除过期数据
				redisTemplate.opsForZSet().removeRangeByScore(queueKey, 0d, rangeMaxScore - msgExpireMillisecond);

				// 取rangeMinScore ~ rangeMaxScore这个分数阶段的数据
				Set<?> zset = redisTemplate.opsForZSet().rangeByScore(queueKey, rangeMinScore, rangeMaxScore);
				if (!zset.isEmpty()) {
					Iterator<?> iterator = zset.iterator();

					Object msg = null;
					while (iterator.hasNext()) {
						// 发送消息
						msg = iterator.next();
						AbstractMsgUtil.sendDelayMessage(bussinessType, bussinessId, msg);
					}

					if (msg != null) {
						// 最后一条消息的分数
						try {
							Double lastMsgScore = redisTemplate.opsForZSet().score(queueKey, msg);
							rangeMinScore = lastMsgScore + 1;// 从最后一条消息之后开始下一轮
							lastTime = new Date().getTime();
						} catch (Exception e) {
							log.error(e.getLocalizedMessage(), e);
						}
					}

					continue;// 直接开始下一轮
				}
				sleep();
			} catch (Exception e) {
				log.error(e.getLocalizedMessage(), e);
			}
		}
	}

	private void sleep() {
		try {
			Thread.sleep(100);
		} catch (InterruptedException e) {
			log.error(e.getLocalizedMessage(), e);
		}
	}
}
